package com.lzwk.tableSql.hive;

import com.lzwk.utils.HiveUtil;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Author: CC
 * @Date: 2022/2/18 15:51
 */
public class HiveSink {
    public static void odsEarthLog(StreamTableEnvironment tableEnv, String tableName, String tablePath) {
        String sql = "create table if not exists warehouse." + tableName + "(" +
                "            `pre_spm` STRING,\n" +
                "            `ref` STRING,\n" +
                "            `ts` TIMESTAMP,\n" +
                "            `ua` STRING,\n" +
                "            `url` STRING,\n" +
                "            `url_params` MAP<STRING, STRING>,\n" +
                "            `spm` STRING,\n" +
                "            `tor_id` STRING,\n" +
                "            `sid` STRING,\n" +
                "            `uid` STRING,\n" +
                "            `rts` STRING,\n" +
                "            `oid` STRING,\n" +
                "            `evt` STRING,\n" +
                "            `evt_params` MAP<STRING, STRING>,\n" +
                "            `ip` STRING,\n" +
                "            `tsk` STRING)\n"
                + HiveUtil.getFlink2HiveDDL(tablePath) + ")";
        System.out.println("HiveSink>>>>>>>>"+sql);
        tableEnv.executeSql(sql);
    }

    public static void dwdEarthLog(StreamTableEnvironment tableEnv, String tableName, String tablePath) {
        String sql = "create table if not exists warehouse." + tableName + "(" +
                "        `uid` STRING,\n" +
                "        `spm` STRING,\n" +
                "        `evt` STRING,\n" +
                "        `evt_params` MAP<STRING, STRING>,\n" +
                "        `ts` TIMESTAMP,\n" +
                "        `rts` STRING,\n" +
                "        `sid` STRING,\n" +
                "        `tor_id` STRING,\n" +
                "        `tsk` STRING,\n" +
                "        `url_pat` STRING,\n" +
                "        `url` STRING,\n" +
                "        `url_params` MAP<STRING, STRING>,\n" +
                "        `spm0` STRING,\n" +
                "        `spm1` STRING,\n" +
                "        `spm2` STRING,\n" +
                "        `spm3` STRING)\n"
                + HiveUtil.getFlink2HiveDDL(tablePath) + ")";
        System.out.println("HiveSink>>>>>>>>"+sql);
        tableEnv.executeSql(sql);
    }
    public static void dirtyEarthLog(StreamTableEnvironment tableEnv, String tableName, String tablePath) {
        String sql = "create table if not exists warehouse." + tableName + "(" +
                "            `@timestamp` STRING," +
                "            `@filepath` STRING," +
                "            `log` STRING" +
                ") " + HiveUtil.getFlink2HiveDDL(tablePath) + ")";
        System.out.println("HiveSink>>>>>>>>"+sql);
        tableEnv.executeSql(sql);
    }
}
